package com.dukk.espark.apps;

import com.alibaba.fastjson.JSONObject;
import com.dukk.espark.core.Engine;
import com.dukk.espark.core.SparkBase;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.collection.mutable.WrappedArray;

import java.util.ArrayList;
import java.util.List;

/**
 * Dataset 常用方法操作示例
 * @version 2020-11-09
 * @author shengshi_feiyang@yeah.net
 */
@Engine(name = "DatasetApp")
public class DatasetApp extends SparkBase {

    @Override
    public void doAi(JSONObject params) {

        //示例一 加载全国省份 json数据
        Dataset<Row> jsonDataset = sparkSession.read().json(configKit.getEsparkFilePath());
        jsonDataset.printSchema();
        jsonDataset.show();
//        +--------------------+-----------+-----------------+
//        |            features|       name|             type|
//        +--------------------+-----------+-----------------+
//       |[[[[[[[116.812128...|全国_origin|FeatureCollection|
//      +--------------------+-----------+-----------------+


        //示例二 select 操作
        Dataset<Row> featuresDataset = jsonDataset.select("features");
        featuresDataset.printSchema();
        featuresDataset.show();

        //示例三 flatMap操作
        //解析features全国省份数据,将features.properties.name 解析出来
        Dataset<Province> provinceDataset = featuresDataset.flatMap((FlatMapFunction<Row, Province>)row->{

            List<Province> provinceList = new ArrayList<>();
            WrappedArray features = (WrappedArray)row.getAs("features");

            for(int i=0; i<features.size(); i++){

                //row 操作
                Row data = (Row)features.apply(i);
                //获取row scheme
                StructType structType = data.schema();
                StructField structFields[] = structType.fields();
                // ... 操作scheme

                Row properties = data.getAs("properties");
                String name = properties.getAs("name");

                Province province = new Province();
                province.setProvince(name);
                provinceList.add(province);
            }

            return provinceList.iterator();
        }, Encoders.bean(Province.class));

        provinceDataset.show();
        //+--------------+
        //|      province|
        //+--------------+
        //|        北京市|
        //|        天津市|
        //|        河北省|
        //|        山西省|
        //|  内蒙古自治区|
        //|        辽宁省|
        //|        吉林省|
        //|      黑龙江省|
        //|        上海市|
        //|        江苏省|
        //|        浙江省|
        //|        安徽省|
        //|        福建省|
        //|        江西省|
        //|        山东省|
        //|        河南省|
        //|        湖北省|
        //|        湖南省|
        //|        广东省|
        //|广西壮族自治区|
        //+--------------+
        //only showing top 20 rows


    }
}
